package com.twoandone.tms.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.twoandone.tms.realtime.beans.TableProcess;
import com.twoandone.tms.realtime.common.TmsConfig;
import com.twoandone.tms.realtime.utils.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Collection;
import java.util.Map;
import java.util.Set;

/**
 * @packageName:com.twoandone.tms.realtime.app.func
 * @ClassName:DimPhoenixSink
 * @Description:
 * @author:
 * @date 2023/3/10 8:35
 **/
public class DimPhoenixSink implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        String sinkTable = jsonObj.getString("sink_table");
//        upsert into KU.tableName （列名） value(列值)
//        指针
//        这里把sink——type删掉了
        jsonObj.remove("sink_type");
        jsonObj.remove("sink_table");
        Set<String> keySet = jsonObj.keySet();
        Collection<Object> values = jsonObj.values();

        String key = StringUtils.join(keySet, ",");
//        因为值都是varchar类型，所以
        String value = StringUtils.join(values, "','");
        String sql = "upsert into " + TmsConfig.PHOENIX_SCHEMA + "." + sinkTable + " (" + key + ") values ('" + value + "')";
        System.out.println("upsert = " + sql);
        PhoenixUtil.excuteSql(sql);
    }
}
